package com.jeffse.light.stream.app.controller;

import com.jeffse.light.stream.app.dataSource.CustomDataSource;
import com.jeffse.light.stream.app.function.*;
import com.jeffse.light.stream.app.function.union.UnionOutPutProcessFunction;
import com.jeffse.light.stream.app.function.union.UnionProcessFunction;
import com.jeffse.light.stream.app.listener.MyPulsarListener;
import com.jeffse.light.stream.app.pojo.AvgTemperature;
import com.jeffse.light.stream.app.pojo.Temperature;
import com.jeffse.light.stream.app.pojo.User;
import com.jeffse.light.stream.core.client.StreamClient;
import com.jeffse.light.stream.core.connector.pulsar.PulsarSource;
import com.jeffse.light.stream.core.data.backend.LocalBackends;
import com.jeffse.light.stream.core.data.context.LightStreamEnv;
import com.jeffse.light.stream.core.data.context.OperatorContext;
import com.jeffse.light.stream.core.data.operator.OutPutStreamOperator;
import com.jeffse.light.stream.core.data.stream.DataStream;
import com.jeffse.light.stream.core.pojo.OutTag;
import com.jeffse.light.stream.core.pojo.WaterMark;
import com.jeffse.light.stream.core.utils.ThreadUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.Serializable;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryMXBean;
import java.lang.management.ThreadInfo;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

@RestController
@RequestMapping("/stream")
@Slf4j
public class SubmitStreamController implements Serializable {

    private StreamClient streamClient = StreamClient.getClient().setPort(9910);

    /**
     * isRunning 不能使用，使用后会报 SubmitStreamController 没有序列化
     */
//    private Boolean isRunning = true;

    @RequestMapping("/submit")
    public void submit() throws Exception {
        DataStream dataStream = new DataStream("测试1");
        StreamClient streamClient = StreamClient.getClient().setPort(9910);
        dataStream.process(new ProcessAFunction());
        dataStream.process(new ProcessBFunction());
        dataStream.map((value)-> value + " map ");
        dataStream.flatMap((value, out) -> {
            String[] arr = (value + "").split("_");
            for (String str: arr) {
                out.collect(str);
            }
        });
        dataStream.addSink((value, ctx)->{
            log.error("数据源："+ ((DataStream)ctx).getDataStreamName() + "   "+ value + "");
            log.error("------------ 活动的线程 ： " + ThreadUtils.getActiveCount());
        });
        dataStream.addSource(ctx -> {
            // 这里可以使用 dataStream 的属性，但是不能使用controller 的属性
            while(dataStream.getIsRunning()){
                Thread.sleep(3000);
                ctx.collect("aaa_bbb");
            }
        });
        streamClient.start(dataStream);
    }

    @RequestMapping("/submit2")
    public void submit2() throws Exception {
        DataStream dataStream = new DataStream("测试2");
        dataStream.process(new ProcessAFunction());

        dataStream.process(new ProcessBFunction());

        dataStream.map((value)-> value + " map ");
        dataStream.flatMap((value, out) -> {
            String[] arr = (value + "").split("_");
            for (String str: arr) {
                out.collect(str);
            }
        });
        dataStream.addSink((value, ctx)->{
            log.error("数据源："+ ((DataStream)ctx).getDataStreamName() + "   "+ value + "");
            log.error("------------ 活动的线程 ： " + ThreadUtils.getActiveCount());
        });
        dataStream.addSource(new CustomDataSource());
        streamClient.start(dataStream);
    }

    /**
     * 远程提交
     * @throws Exception
     */
    @RequestMapping("/submitKeyedByProcess")
    public void submit3() throws Exception {
        DataStream<String, User, User> dataStream = new DataStream("测试3");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        dataStream.keyedBy(value -> value.getName()).process(new ListFunction());
        log.error("--------------------------------开始处理数据流--------------------------------------");
        dataStream.addSink((value, ctx)->{
            log.error(((DataStream)ctx).getDataStreamName() + ": " + value + "");
            log.error("------------ 活动的线程 ： " + ThreadUtils.getActiveCount());
        });
        dataStream.addSource(ctx -> {
            Integer idx = 0;
            while(dataStream.getIsRunning()){
                Thread.sleep(3000);
                ctx.collect(new User("yiyi", "女", 0 + idx));
                ctx.collect(new User("dad", "男", 34 + idx));
                idx++;
            }
        });

        streamClient.start(dataStream);
    }

    /**
     * 远程提交
     * @throws Exception
     */
    @RequestMapping("/submitWindowSlide")
    public void submit4() throws Exception {
        DataStream<String, Temperature, AvgTemperature> dataStream = new DataStream("test");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        dataStream.setWaterMark(value ->  new WaterMark(value.getTimestamp()));
        dataStream.keyedBy(value -> value.getCity()).process(new WindowSlideFunction());
        dataStream.addSink((value, ctx)->{
            log.error(value.toString());
            log.error("------------ 活动的线程 ： " + ThreadUtils.getActiveCount());
        });
        dataStream.addSource(ctx -> {
            Random r = new Random();
            while(dataStream.getIsRunning()){
                Thread.sleep(1000);
                ctx.collect(new Temperature("上海", r.nextDouble() * 40, System.currentTimeMillis()));
            }
        });
        log.error("--------------------------------开始处理数据流--------------------------------------");
        streamClient.start(dataStream);
    }

    /**
     * 远程提交
     * @throws Exception
     */
    @RequestMapping("/submitOutputSide")
    public void submit5() throws Exception {
        DataStream<String, Temperature, AvgTemperature> dataStream = new DataStream("test");
        Boolean isRunning = true;
        OutTag outTag1 = new OutTag("上海");
        OutTag outTag2 = new OutTag("郑州");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        dataStream.addSource(ctx -> {
            Random r = new Random();
            while(isRunning){
                Thread.sleep(1000);
                List<Temperature> temperatureList = new ArrayList<>();
                temperatureList.add(new Temperature("上海", r.nextDouble() * 40, System.currentTimeMillis()));
                temperatureList.add(new Temperature("郑州", r.nextDouble() * 47 - 8, System.currentTimeMillis()));
                ctx.collect(temperatureList);
            }
        });
        dataStream.flatMap((value, out)->{
            for(Temperature temperature: (List<Temperature>)value){
                if(temperature.getCity().equals("郑州")){
                    out.collect(temperature);
                }else{
                    out.collect(temperature);
                }
            }
        });

        OperatorContext<String, Temperature , AvgTemperature> outPutStreamOperator1 = dataStream.getSideOutput(outTag1);
        OperatorContext<String, Temperature, AvgTemperature> outPutStreamOperator2 = dataStream.getSideOutput(outTag2);

        dataStream.process(new OutPutSideProcessFunction(outTag1, outTag2));
        outPutStreamOperator1.setWaterMark(value ->  new WaterMark(value.getTimestamp()));
        outPutStreamOperator1.keyedBy(value -> value.getCity()).process(new OutPutSideFunction());

        dataStream.addSink((value, ctx) -> {
            log.error("2--------------" + value.toString());
            log.error("------------ 活动的线程 ： " + ThreadUtils.getActiveCount());
        });

        log.error("--------------------------------开始处理数据流--------------------------------------");
//        dataStream.run();
        streamClient.start(dataStream);
    }

    /**
     * 远程提交
     * @throws Exception
     */
    @RequestMapping("/submitUnion")
    public void submit6() throws Exception {
        DataStream<String, Temperature, String> dataStream = new DataStream("test");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        OutTag outTag1 = new OutTag("上海");
        OutTag outTag2 = new OutTag("郑州");
        OutTag outTag3 = new OutTag("杭州");
        Boolean isRunning = true;
        dataStream.addSource(ctx -> {
            Random r = new Random();
            while(isRunning){

                ctx.collect(new Temperature("上海", r.nextDouble() * 40, System.currentTimeMillis()));
                ctx.collect(new Temperature("郑州", r.nextDouble() * 40, System.currentTimeMillis()));
                ctx.collect(new Temperature("杭州", r.nextDouble() * 40, System.currentTimeMillis()));
                Thread.sleep(1000);
            }
        });

        OperatorContext<String, Temperature , AvgTemperature> outPutStreamOperator1 = dataStream.getSideOutput(outTag1);
        OperatorContext<String, Temperature, AvgTemperature> outPutStreamOperator2 = dataStream.getSideOutput(outTag2);
        OperatorContext<String, Temperature, AvgTemperature> outPutStreamOperator3 = dataStream.getSideOutput(outTag3);
//
        dataStream.process(new UnionOutPutProcessFunction(outTag1, outTag2, outTag3));
        outPutStreamOperator1.setWaterMark(value ->  new WaterMark(value.getTimestamp()));
        outPutStreamOperator1.keyedBy(value -> value.getCity()).process(new OutPutSideFunction());

        outPutStreamOperator2.keyedBy(value -> value.getCity()).process(new UnionProcessFunction());

        outPutStreamOperator3.keyedBy(value -> value.getCity()).process(new UnionProcessFunction());

        OutPutStreamOperator outPutStreamOperator = (OutPutStreamOperator) outPutStreamOperator1.union(((OutPutStreamOperator)outPutStreamOperator2).getOutTag()).union(((OutPutStreamOperator)outPutStreamOperator3).getOutTag());
        outPutStreamOperator.map(item-> ((AvgTemperature)item).getCity());
//
        dataStream.addSink((value, ctx) -> {
            log.error("2--------------" + value.toString());
            log.error("------------ 活动的线程 ： " + ThreadUtils.getActiveCount());
        });

        log.error("--------------------------------开始处理数据流--------------------------------------");
//        dataStream.run();
        streamClient.start(dataStream);
    }

    /**
     * 本地运行
     * @throws Exception
     */
    @RequestMapping("/startKeyedByProcess")
    public void start() throws Exception {
        DataStream<String, User, User> dataStream = new DataStream("测试3");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        dataStream.keyedBy(value -> value.getName()).process(new ListFunction());
        log.error("--------------------------------开始处理数据流--------------------------------------");
        dataStream.addSink((value, ctx)->{
            log.error(((DataStream)ctx).getDataStreamName() + ": " + value + "");
        });
        dataStream.addSource(ctx -> {
            Integer idx = 0;
            while(dataStream.getIsRunning()){
                Thread.sleep(3000);
                ctx.collect(new User("yiyi", "女", 0 + idx));
                ctx.collect(new User("dad", "男", 34 + idx));
                idx++;
            }
        });

        dataStream.start();
    }

    /**
     * 本地运行
     * @throws Exception
     */
    @RequestMapping("/startKeyedByProcess2")
    public void start2() throws Exception {
        DataStream<String, User, User> dataStream = new DataStream("测试4");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        dataStream.keyedBy(value -> value.getName()).process(new ListFunction());
        log.error("--------------------------------开始处理数据流--------------------------------------");
        dataStream.addSink((value, ctx)->{
            log.error(((DataStream)ctx).getDataStreamName() + ": " + value + "");
        });
        dataStream.addSource(ctx -> {
            Integer idx = 0;
            while(dataStream.getIsRunning()){
                Thread.sleep(3000);
                ctx.collect(new User("yiyi", "女", 0 + idx));
                ctx.collect(new User("dad", "男", 34 + idx));
                idx++;
            }
        });

        dataStream.start();
    }

    @RequestMapping("/pulsarSource")
    public void pulsarSource() throws Exception {
        DataStream<String, User, User> dataStream = new DataStream<>("pulsarSource");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        log.info("--------------------------------开始处理数据流--------------------------------------");
        dataStream.addSink((value, ctx)->{
            log.info(((DataStream)ctx).getDataStreamName() + ": " + value + "");
        });
        dataStream.addSource(new PulsarSource("pulsar://localhost:6650", "test", "test", MyPulsarListener.class));
//        dataStream.start();
        streamClient.start(dataStream);
    }

    @RequestMapping("/pulsarSource2")
    public void pulsarSource2() throws Exception {
        DataStream<String, User, User> dataStream = new DataStream<>("pulsarSource");
        LightStreamEnv.getENV().setBackends(new LocalBackends());
        log.info("--------------------------------开始处理数据流--------------------------------------");
        dataStream.addSink((value, ctx)->{
            log.info(((DataStream)ctx).getDataStreamName() + ": " + value + "");
        });

        List<String> topics = new ArrayList<>();
        topics.add("test");
        topics.add("test1");
        dataStream.addSource(new PulsarSource("pulsar://localhost:6650", topics, "test", MyPulsarListener.class));
//        dataStream.start();
        streamClient.start(dataStream);
    }

    @RequestMapping("/cancel/{threadName}")
    public void cancel(@PathVariable("threadName") String threadName) throws Exception {
        streamClient.cancel(threadName);
    }

    @RequestMapping("/getThreadsInfo/{threadName}")
    public List<ThreadInfo> getThreadsInfo(@PathVariable("threadName") String threadName) {
        return ThreadUtils.getThreadsInfo(threadName);
    }

    @RequestMapping("/getDeadThreadsInfo/{threadName}")
    public List<ThreadInfo> getDeadThreadsInfo(@PathVariable("threadName") String threadName) {
        return ThreadUtils.getDeadThreadsInfo(threadName);
    }

    @RequestMapping("/getAllThreadsInfo")
    public List<ThreadInfo> getDeadThreadsInfo() {
        return ThreadUtils.getAllThreadsInfo();
    }

    @RequestMapping("/getMemoryInfo")
    public List<String> getMemoryInfo() {
        MemoryMXBean mm=(MemoryMXBean) ManagementFactory.getMemoryMXBean();
        List<String> list = new ArrayList<>();
        list.add("堆内存占用: "+mm.getHeapMemoryUsage());
        list.add("栈内存占用: "+mm.getNonHeapMemoryUsage());
        return list;
    }


    @RequestMapping("/batchExecute")
    public void batchExecute() throws Exception {
        ThreadUtils.batchExecute();
    }
}
